package com.udf.flink.udaf;
/*
 * @author: create by jianglai
 * @version: v1.0
 * @description: com.dtstack.flinksql.udaf
 * @date:2021/3/25
 **/

import org.apache.flink.table.functions.AggregateFunction;

/**
 * @description：TODO
 * @author：jianglai
 * @date：2021/3/25 10:05 下午
 */
public class LastValue  extends AggregateFunction<String,LastValue.acc> {

    @Override
    public String getValue(acc accumulator) {
        return accumulator.finaDate;
    }

    @Override
    public acc createAccumulator() {
        acc acc = new acc();
        acc.finaDate = null;
        acc.time = null;
        return acc;
    }

    public class acc {
        public String finaDate;
        public String time = null;
    }

    public void accumulate(acc acc,Object key, Object value){
        if (null != key && null != value){
            if (acc.time == null || acc.time.compareTo(key.toString()) <= 0) {
                    acc.time = key.toString();
                    acc.finaDate = value.toString();
            }
        }
    }
}

